package kafka

import (
	"github.com/Shopify/sarama"
	"github.com/alecthomas/log4go"
	"sync"
)

type Producer struct {
	producer  sarama.AsyncProducer
	topica     string
	topicb     string
	msgaQ     chan *sarama.ProducerMessage
	msgbQ     chan *sarama.ProducerMessage
	wg        sync.WaitGroup
	closeChan chan struct{}
}

// Close 关闭
func (p *Producer) Close() error {
	close(p.closeChan)
	log4go.Warn("[producer] is quiting")
	p.wg.Wait()
	log4go.Warn("[producer] quit over")
	return p.producer.Close()
}

func (p *Producer) SetMsgaQ(msgQmake chan *sarama.ProducerMessage) {
	p.msgaQ = msgQmake
}
func (p *Producer) SetMsgbQ(msgQmake chan *sarama.ProducerMessage) {
	p.msgbQ = msgQmake
}
func (p *Producer) GetTopicA() string{
	return p.topica
}
func (p *Producer) GetTopicB() string{
	return p.topicb
}

// Run 运行
func (p *Producer) Run() {

	p.wg.Add(1)
	go func() {
		defer p.wg.Done()

	LOOP:
		for {
			select {
			case m := <-p.msgaQ:
				p.producer.Input() <- m
			case err := <-p.producer.Errors():
				if nil != err && nil != err.Msg {
					log4go.Error("[producer] err=[%s] topic=[%s] key=[%s] val=[%s]", err.Error(), err.Msg.Topic, err.Msg.Key, err.Msg.Value)
				}
			case <-p.closeChan:
				break LOOP
			}
			select {
			case m := <-p.msgbQ:
				p.producer.Input() <- m
			case err := <-p.producer.Errors():
				if nil != err && nil != err.Msg {
					log4go.Error("[producer] err=[%s] topic=[%s] key=[%s] val=[%s]", err.Error(), err.Msg.Topic, err.Msg.Key, err.Msg.Value)
				}
			case <-p.closeChan:
				break LOOP
			}

		}
	}()

	for hasTask := true; hasTask; {
		select {
		case m := <-p.msgaQ:
			p.producer.Input() <- m
		default:
			hasTask = false
		}
		select {
		case m := <-p.msgbQ:
			p.producer.Input() <- m
		default:
			hasTask = false
		}
	}

}

